package com.boyi.center.task;

import com.boyi.shared.kafka.KafkaResult;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @author tingwen
 */
public abstract class AbstractKafkaTask implements KafkaTask {
	@Autowired
	private KafkaTemplate kafkaTemplate;

	final static int FAILURE_TIME = 5;

	@Override
	public void taskPublic(ProducerRecord<String, KafkaResult<Object>> record) {
		ListenableFuture future = kafkaTemplate.send(record);
		future.addCallback(new ListenableFutureCallback() {
			@Override
			public void onFailure(Throwable ex) {
				handleFailure(record);
			}

			@Override
			public void onSuccess(Object result) {
				handleSuccess(result);
			}
		});
	}
}
